package com.momo.apachedoris.service.unique;

import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.github.yitter.idgen.YitIdHelper;
import com.momo.apachedoris.dao.vo.DorisConnectionVO;
import com.momo.apachedoris.dao.vo.DorisSendDataVO;
import com.momo.apachedoris.dao.vo.StreamLoadVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.FileEntity;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.DefaultRedirectStrategy;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.springframework.http.HttpHeaders;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;

/**
 * @Author: Jie Li
 * @Date 2022-04-10 11:58
 * @Description: 不是全字段导入需要指定columns对应关系，
 * 而且没有导入的数据会被置为null,而不是空字符串
 * @Version: 1.0
 * <p>Copyright: Copyright (c) 2022</p >
 **/
@Service
@Slf4j
public class DorisStreamLoadService {

    public StreamLoadVO sendData(DorisSendDataVO sendDataVO) throws IOException {
        DorisConnectionVO connection = sendDataVO.getConnection();
        final String loadUrl = String.format("http://%s:%s/api/%s/%s/_stream_load",
                connection.getHost(),
                connection.getHttpPort(),
                connection.getDb(),
                sendDataVO.getTable());

        final HttpClientBuilder httpClientBuilder = HttpClients
                .custom()
                .setRedirectStrategy(new DefaultRedirectStrategy() {
                    @Override
                    protected boolean isRedirectable(String method) {
                        return true;
                    }
                });

        CloseableHttpClient client = httpClientBuilder.build();
        HttpPut put = new HttpPut(loadUrl);

        if (StringUtils.isNotBlank(sendDataVO.getContent())) {
            StringEntity entity = new StringEntity(sendDataVO.getContent(), "UTF-8");
            put.setEntity(entity);
        } else if (sendDataVO.getFile() != null) {
            FileEntity entity = new FileEntity(sendDataVO.getFile());
            put.setEntity(entity);
        }

        put.setHeader(HttpHeaders.EXPECT, connection.getExpect());
        put.setHeader(HttpHeaders.AUTHORIZATION, connection.basicAuthHeader());
        // the label header is optional, not necessary
        // use label header can ensure at most once semantics
        if (StringUtils.isBlank(sendDataVO.getLabel())) {
            sendDataVO.setLabel(String.valueOf(YitIdHelper.nextId()));
        }
        if (StringUtils.isNotBlank(sendDataVO.getColumn_separator())) {
            put.setHeader("column_separator", sendDataVO.getColumn_separator());
        }
        if (StringUtils.isNotBlank(sendDataVO.getLine_delimiter())) {
            put.setHeader("line_delimiter", sendDataVO.getLine_delimiter());
        }
        if (StringUtils.isNotBlank(sendDataVO.getColumns())) {
            put.setHeader("columns", sendDataVO.getColumns());
        }
        if (StringUtils.isNotBlank(sendDataVO.getFormat())) {
            put.setHeader("format", sendDataVO.getFormat());
        }
        if (StringUtils.isNotBlank(sendDataVO.getJsonpaths())) {
            put.setHeader("jsonpaths", sendDataVO.getJsonpaths());
        }
        if (StringUtils.isNotBlank(sendDataVO.getJsonpaths())) {
            put.setHeader("jsonpaths", sendDataVO.getJsonpaths());
        }
        if (sendDataVO.getStrip_outer_array() != null) {
            put.setHeader("strip_outer_array", String.valueOf(sendDataVO.getStrip_outer_array()));
        }
        if (sendDataVO.getFuzzy_parse() != null) {
            put.setHeader("fuzzy_parse", String.valueOf(sendDataVO.getFuzzy_parse()));
        }
        if (sendDataVO.getNum_as_string() != null) {
            put.setHeader("num_as_string", String.valueOf(sendDataVO.getNum_as_string()));
        }
        if (StringUtils.isNotBlank(sendDataVO.getMerge_type())) {
            put.setHeader("merge_type", sendDataVO.getMerge_type());
        }
        if (StringUtils.isNotBlank(sendDataVO.getSequence_col())) {
            put.setHeader("function_column.sequence_col", sendDataVO.getSequence_col());
        }
        if (StringUtils.isNotBlank(sendDataVO.getDelete()) && DorisSendDataVO.MERGE.equals(sendDataVO.getMerge_type()) ||
                StringUtils.isNotBlank(sendDataVO.getDelete()) && DorisSendDataVO.DELETE.equals(sendDataVO.getMerge_type())) {
            put.setHeader("delete", sendDataVO.getDelete());
        }

        put.setHeader("label", sendDataVO.getLabel());

        try (CloseableHttpResponse response = client.execute(put)) {
            String loadResult = "";
            if (response.getEntity() != null) {
                loadResult = EntityUtils.toString(response.getEntity());
            }
            final int statusCode = response.getStatusLine().getStatusCode();
            // statusCode 200 just indicates that starrocks be service is ok, not stream load
            // you should see the output content to find whether stream load is success
            if (statusCode != 200) {
                log.error(String.format("Stream load failed, statusCode=%s load result=%s", statusCode, loadResult));
                throw new IOException();
            }
            StreamLoadVO streamLoadVO = JSON.parseObject(loadResult, new TypeReference<StreamLoadVO>() {
            });
            if (StringUtils.isNotBlank(streamLoadVO.getStatus()) && !streamLoadVO.getStatus().toLowerCase().contains("success")) {
                log.error("streamload error => table= {} , label={} , Message={} , ErrorURL={} ", sendDataVO.getTable(), sendDataVO.getLabel(), streamLoadVO.getMessage(), streamLoadVO.getErrorURL());
            }

            streamLoadVO.setLabel(sendDataVO.getLabel());
            return streamLoadVO;
        }
    }

    public static void main(String[] args) throws UnsupportedEncodingException {
        String ss = URLEncoder.encode("张三", "UTF-8");
        System.out.println(ss);
    }
}
